package com.chinastock.serializer;

import com.alibaba.fastjson.JSON;
import com.chinastock.log.LogName;
import com.chinastock.model.EcpMessage;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.context.ContextView;

import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.Function;

/**
 * @author zhangfan.it
 * @date 2021/2/4
 */
public class JsonDecoder implements Decoder<Object> {
    private static final Logger log = LoggerFactory.getLogger(LogName.ECP);

    @Override
    public boolean canDecode(ResolvableType elementType, MimeType mimeType) {
        assert mimeType != null;
        return mimeType.includes(MimeTypeUtils.APPLICATION_JSON);
    }

    @Override
    public Flux<Object> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) {
        return Flux.from(inputStream).handle(((dataBuffer, sink) -> {
            try {
                Object value = parse(dataBuffer, elementType);
                if (value != null) {
                    sink.next(value);
                }
            }
            catch (Exception ex) {
                sink.error(ex);
            }
        }));
    }

    @Override
    public Mono<Object> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) {
        return DataBufferUtils.join(inputStream).map(dataBuffer-> parse(dataBuffer,elementType));
    }

    @Override
    public List<MimeType> getDecodableMimeTypes() {
        return Collections.singletonList(MimeTypeUtils.APPLICATION_JSON);
    }

    private Object parse(DataBuffer dataBuffer,ResolvableType elementType) {
        try {
            InputStream inputStream = dataBuffer.asInputStream();
            byte[] bytes = new byte[inputStream.available()];
            inputStream.read(bytes);
            String text = new String(bytes, StandardCharsets.UTF_8);
            return JSON.parseObject(text, elementType.getType());
        } catch (IOException e) {
            Mono.deferContextual((Function<ContextView, Mono<ServerHttpRequest>>) contextView -> Mono.just(contextView.get(ServerHttpRequest.class)))
                    .doOnSuccess(serverHttpRequest -> {
                        EcpMessage message= new EcpMessage();
                        message.setUrl(serverHttpRequest.getURI().getPath());
                        message.setException(e);
                        log.error(message.toString());
                    });
            return null;
        } finally {
            DataBufferUtils.release(dataBuffer);
        }
    }
}
